package net.xuele.learn.flink.book;

import net.xuele.learn.flink.book.utils.SensorReading;
import net.xuele.learn.flink.book.utils.SensorSource;
import net.xuele.learn.flink.book.utils.SensorTimeAssigner;

import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkGeneratorSupplier;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
/**
 * @Author patrick
 * @Date 2023/7/5 10:11
 * @Description
 */
public class AverageSensorReadings {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        env.getConfig().setAutoWatermarkInterval(1000L);


        DataStream<SensorReading> sensorData = env.addSource(new SensorSource())
                // 分配事件事件所需的时间搓和水位线
                .assignTimestampsAndWatermarks(new WatermarkStrategy<SensorReading>() {
                    @Override
                    public WatermarkGenerator<SensorReading> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                        // 这是新版本的watermark方法
                        return null;
                    }
                })
                .assignTimestampsAndWatermarks(new SensorTimeAssigner());

        DataStream<SensorReading> avgTemp = sensorData
                // convert Fahrenheit to Celsius using and inlined map function 华氏度转摄氏度
                .map(r -> new SensorReading(r.id, r.timestamp, (r.temperature - 32) * (5.0 / 9.0)))
                // organize stream by sensor
                .keyBy(r -> r.id)
                // group readings in 1 second windows  ———— 滚动窗口
                .timeWindow(Time.seconds(1))
                // compute average temperature using a user-defined function
                // 使用自定义函数计算每个窗口的平均温度，窗口之后接的方法都是处理单个窗口中累计数据用的，而不是所有的！！！
                .apply(new TemperatureAverager());

        // print result stream to standard out
        avgTemp.writeAsText("outfile");

        // execute application
        /**
         * 构建完的计划会被转成 JobGraph 并提交至 JobManager 执行
         * 可以是本地JobManager，也可能会将其发送到远程 JobManager 上。
         * 如果是后者，除 JobGraph 之外，我们还要同时提供包含应用所需全部类和依赖的 JAR 包。
         */
        env.execute("Compute average sensor temperature");
    }


    /**
     * User-defined WindowFunction to compute the average temperature of SensorReadings
     */
    public static class TemperatureAverager implements WindowFunction<SensorReading, SensorReading, String, TimeWindow> {

        /**
         * apply() is invoked once for each window.
         *
         * @param sensorId the key (sensorId) of the window
         * @param window   meta data for the window
         * @param input    an iterable over the collected sensor readings that were assigned to the window
         * @param out      a collector to emit results from the function
         */
        @Override
        public void apply(String sensorId, TimeWindow window, Iterable<SensorReading> input, Collector<SensorReading> out) {

            // compute the average temperature
            int cnt = 0;
            double sum = 0.0;
            for (SensorReading r : input) {
                cnt++;
                sum += r.temperature;
            }
            double avgTemp = sum / cnt;
            // emit a SensorReading with the average temperature
            // 最终被采集的数据
            out.collect(new SensorReading(sensorId, window.getEnd(), avgTemp));
        }
    }
}
